package com.atbeijing.bigdata.spark.mytest.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable


/**
 * Map[品类，HotCategory]
 *
 */
object Top10_3 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("top")
    val sc = new SparkContext(conf)

    //先按照点击数排名，靠前的就排名高；如果点击数相同，再比较下单数；下单数再相同，就比较支付数。
    val file: RDD[String] = sc.textFile("data/user_visit_action.txt")

    val obj: RDD[UserVisitAction] = file.map(line => {
      val datas = line.split("_")
      UserVisitAction(
        datas(0),
        datas(1).toLong,
        datas(2),
        datas(3).toLong,
        datas(4),
        datas(5),
        datas(6).toLong,
        datas(7).toLong,
        datas(8),
        datas(9),
        datas(10),
        datas(11),
        datas(12).toLong
      )
    })

    //每个页面访问次数
    val map1: Map[Long, Int] = obj.map(o => {
      (o.page_id, 1)
    }).reduceByKey(_ + _).collect().toMap

    obj.cache()

    val session: RDD[(String, Iterable[UserVisitAction])] = obj.groupBy(o => o.session_id)
    //每个用户访问的页面
    //(sessionId, List[(页面id-页面id, 1)])
    val mapDatas: RDD[(String, List[(String, Int)])] = session.mapValues(v => {
      val actions: List[UserVisitAction] = v.toList.sortBy(o => o.action_time)
      val ids: List[Long] = actions.map(a => a.page_id)
      val flowids: List[(Long, Long)] = ids.zip(ids.tail)
      // 【（1-2，1）， （2-3，1）】
      flowids.map {
        case (id1, id2) => {
          (id1 + "-" + id2, 1)
        }
      }
    })
    // RDD[(String, Int)] => ("a", 1), ("a", 2)
    // RDD[List[(String, Int)]] => List( ("a",1) ), List(("a", 2))
    val flatDatas: RDD[(String, Int)] = mapDatas.map(_._2).flatMap(list=>list)
    //所有用户的数据(页面id-页面id, sum)
    val pageFlowSum: RDD[(String, Int)] = flatDatas.reduceByKey(_ + _)


    //分子
    pageFlowSum.foreach {
      //pageflow某一种跳转A->B      sum次数
      case ( pageflow, sum ) => {
        val ids = pageflow.split("-")
        //ids(0)跳转前的页面id
        val cnt = map1.getOrElse(ids(0).toLong, 1)
        //A->B跳转的次数/A页面访问总次数就是A->B的转换率
        println(s"页面跳转【${pageflow}】的转换率为 = " + (sum.toDouble/cnt))
      }
    }
  }

  case class UserVisitAction(
                              date: String,//用户点击行为的日期
                              user_id: Long,//用户的ID
                              session_id: String,//Session的ID
                              page_id: Long,//某个页面的ID
                              action_time: String,//动作的时间点
                              search_keyword: String,//用户搜索的关键词
                              click_category_id: Long,//某一个商品品类的ID
                              click_product_id: Long,//某一个商品的ID
                              order_category_ids: String,//一次订单中所有品类的ID集合
                              order_product_ids: String,//一次订单中所有商品的ID集合
                              pay_category_ids: String,//一次支付中所有品类的ID集合
                              pay_product_ids: String,//一次支付中所有商品的ID集合
                              city_id: Long//城市 id
                            )
}
